…, awaitable transfer facade (#4865)

The complete F3/Cell payload layer the Client API execution modes build
on, plus the design revision that documents it. Upper layers (executor
backends, trainer engine) consume one primitive:

```python
waiter = downloader.get_waiter()
outcome = waiter.wait(timeout=...)   # returns == delivered
```

*(This PR ships the primitive and the truth model behind it; production
callers of `get_waiter()` arrive with the trainer-engine and
executor-backend PRs. Nothing user-visible changes yet.)*

### How a huge-model send flows: before and after

**Before this PR — served is assumed delivered.** The sender's books say
SUCCESS when the last chunk leaves; whether the receiver could actually
store the model is unknowable, and the terminal FINISHED status counts
failed receivers too:

```mermaid
sequenceDiagram
    participant S as sender cell (holds 70GB model)
    participant R as receiver cell

    Note over S: flare.send() registers refs, then waits on ad-hoc<br/>machinery: progress messages, timeout guesses, ack grace.
    S->>R: small control message (ref ids, no payload)

    loop receiver-driven chunk pulls
        R->>S: request(R1, state)
        S-->>R: reply(chunk, next state)
    end

    R->>S: request(R1, final state)
    S-->>R: reply(EOF)
    Note over S: sender counts this receiver as SUCCESS<br/>the moment the last chunk is served.

    Note over R: receiver stores the model (e.g. disk write).<br/>If this fails, nobody tells the sender.

    Note over S: transaction FINISHED means receiver COUNT reached,<br/>counting failed receivers too.<br/>No queryable verdict exists afterward.
    Note over S,R: consequences: the sender frees the model or exits its<br/>subprocess on a guess. A receiver whose store failed is<br/>invisible: the workflow proceeds without the model (silent truncation).
```

**With this PR — served ≠ delivered; the receiver's own stored/failed
report decides.** Every receiver's outcome is knowable, bounded in time,
and queryable afterward:

```mermaid
sequenceDiagram
    participant S as sender cell (holds 70GB model)
    participant R as receiver cell

    Note over S: sender registers transaction T0 with refs R1..Rn.<br/>The model stays in sender memory - only refs travel.<br/>waiter = get_transfer_waiter(T0) is the facade this PR adds.
    S->>R: small control message (ref ids, no payload)

    loop receiver-driven chunk pulls (MBs per chunk)
        R->>S: request(R1, state, CONFIRM_CAPABLE)
        S-->>R: reply(chunk, next state)
    end

    R->>S: request(R1, final state, CONFIRM_CAPABLE)
    S-->>R: reply(EOF, CONFIRM_EXPECTED)
    Note over S: sender notes: all bytes served to this receiver,<br/>outcome pending the receiver's confirmation.<br/>Not yet counted as delivered.

    Note over R: receiver stores the assembled model<br/>(download_completed, e.g. write the file to disk).<br/>This step can fail even though every byte arrived.
    R->>S: CONFIRM: stored OK / store failed (fire and forget)

    Note over S: sender records what the receiver reported.<br/>stored OK: counted as delivered.<br/>store failed: counted as failed, despite the served EOF.
    Note over S: when every expected receiver is counted, the monitor closes T0:<br/>callbacks run, 70GB source release attempted (errors logged), THEN the verdict is recorded<br/>(kept 30 min) and waiter.wait() returns - fully settled.

    Note over S,R: bounds on the sender's 5s monitor tick. Budgets are opt-in,<br/>set per transaction or via config vars - acquire needs declared receiver ids.<br/>receiver never pulls: failed at the acquire budget.<br/>receiver goes silent or its CONFIRM is lost: failed at the idle budget.<br/>either budget unset: the transaction TTL remains the backstop.<br/>old-version receiver (no CONFIRM_CAPABLE): counted at EOF, exactly today's behavior.
```

### The trade-off, stated plainly

Holding the source: the sender keeps the model longer **only on the
happy path** (+ receiver store time + one confirm hop) — exactly the
window in which the old code had already freed a model whose delivery
was unproven. In the degraded cases the hold is *shorter* than before:

| Path | Before | After |
|---|---|---|
| Happy path | freed when last chunk **served** | + receiver store +
confirm hop |
| Receiver stalls mid-pull | pinned until full tx TTL | freed at the
**idle budget** (sooner) |
| Receiver never pulls | pinned until TTL | freed at the **acquire
budget** (sooner) |
| Absolute worst case | TTL | TTL (unchanged backstop) |

*The budget rows apply when budgets are configured (opt-in, per
transaction or via `streaming_receiver_acquire_timeout` /
`streaming_receiver_idle_timeout`; the acquire budget additionally
requires declared receiver identities). Unconfigured deployments keep
exactly the before-column TTL behavior.*

Network hiccups: chunk/EOF loss is handled as before (retries, tombstone
replay). The one new must-arrive message is the CONFIRM; if it is lost,
the receiver is counted FAILED despite a good store — a **false
negative**, bounded by the idle budget, healed by an idempotent retry.
This is the Two Generals bound: some message is always last and
unconfirmed, so a protocol only chooses which way it errs. Before, we
erred toward invisible false success (silent truncation); now we err
toward visible, bounded, retryable false failure.
Corruption-by-assumption is no longer a possible outcome.

**Outcome-recording hardening** (follow-up to #4853's F3-1): **tx_ids
are attempt-scoped and single-use while known** — a retry of the same
logical transfer is a new transaction with a new id, and
`new_transaction` rejects a duplicate id (`ValueError`) while it is
live, still settling, its receipt is retained (`TX_OUTCOME_TTL`), or a
drain-timeout leak from a previous attempt is still in flight. The
default fresh uuid never collides; a caller-supplied id becomes reusable
only after its receipt expired AND every old operation exited. The
stable cross-attempt identity is the application-level transfer id
carried in the caller's metadata; it never enters this service.
Uniqueness is what makes every emission of a terminated transaction
attributable to it alone — no suppression flags, generation
serialization, or settlement waiting exist (an earlier revision
serialized reused-id generations; review falsified that boundary
repeatedly, so the boundary was deleted at the naming level). Ownership
(`_outcome_owners`) and `_tx_table` registration remain one atomic step,
so a transaction the monitor can terminate always owns its outcome slot,
concurrent same-id constructors resolve to exactly one winner and clean
errors for the losers, and `shutdown()` clears ownership atomically with
table teardown (nothing records after shutdown; the owner guard alone
gates it). In-flight operations (serves, confirms, monitor budget
passes) register with a per-transaction **activity gate** that
settlement closes and drains before snapshotting the outcome — an
operation finishing during termination is counted in the verdict, and
nothing emits against a settled transaction (only a callback hanging
beyond the 60s drain timeout can leak a late emission, loudly logged —
and such a leak keeps its tx_id excluded from registration until the
operation exits, so a leaked emission can never land under a recycled
id); `tx` required in `_record_outcome` (recording is legal only for the
slot owner); deep-frozen `TransferOutcome` (note: `receiver_statuses` is
a `MappingProxyType` — not JSON/pickle-serializable by design; consumers
crossing a boundary materialize with `dict(...)`); exception-guarded
`downloaded_to_*` callbacks.

**Receiver-confirmed completion + retry-aware accounting**: a
confirm-capable receiver's served EOF/ERROR is provisional; the
receiver's fire-and-forget confirmation — carrying its stored-OK /
store-failed truth (a finalization failure confirms FAILED) — finalizes
it. Receiver truth wins (served-EOF + failed store = FAILED — the
disk-offload case); retries overwrite provisional records. Every
confirmation echoes a **per-serve nonce** from the terminal reply,
binding it to its exact serve: a stale confirmation from a previous life
of a reused ref_id is dropped even when the new life has its own pending
serve for the same receiver. Confirmations ride with the download's
`secure` setting. All wire keys optional (both version skews degrade to
today's producer-served semantics); per-process kill-switch
`streaming_receiver_confirm_enabled` disables the wire behavior on
either side without a code revert.

**Declared receiver identities + per-(transfer, receiver) budgets +
quorum**: when `receiver_ids` are declared, completion, the aggregate
outcome, and the quorum are judged against **those identities** — a
status from an unexpected receiver can never complete a transfer a
declared receiver did not get. Budgets (opt-in): a never-pulling
receiver fails at the acquire budget (requires declared identities);
idleness is judged on **transaction-level** activity, so a receiver that
finished one ref and went silent cannot escape a sibling ref's budgets —
with a truth-wins re-check on enforcement. Budget failures resolve the
outcome on a monitor pass instead of the TTL, which also bounds lost
confirmations (fail-closed). `min_receivers`/`quorum_met` give fan-out
workflows the k-of-N surface (a quorum receiver must be a declared
receiver and succeed on every ref); `completed` stays the strict
all-receivers certificate.

**`TransferWaiter` awaitable facade**: event-driven and
**settle-then-resolve** — the outcome is recorded (and waiters released)
only after the callback chain and the source-release attempts complete
(a raising `release()` is logged and cannot block settlement or strand
the waiter), so acting on `wait()` returning can never preempt them.
Never hangs (unknown/expired/shut-down ids resolve immediately; shutdown
releases all waiters). tx_ids are attempt-scoped, so a waiter always
resolves with the verdict of exactly the attempt it named; a retrying
caller acquires a new waiter for the new attempt. Optional
caller-specified linger (FINISHED-gated) preserves the tombstone replay
window; `acquired_receivers()` is the V1 PAYLOAD_ACQUIRED signal;
composes with — never replaces — the existing `DOWNLOAD_COMPLETE_CB`
chain.

**Design rev 2.2 + vocabulary** (folded from the former #4866):
forward-path heartbeat exemption narrowed to the payload-materialization
phase via new `Topic.TASK_PAYLOAD_READY` (heartbeats stay authoritative
during training; diagrams re-validated); `launch_once=False` moves to
launch-scoped tokens; F3 dependency/vocabulary-mapping updates; status
set to Approved. The implementation-plan doc is removed — it was a
point-in-time PR decomposition that execution has already deviated from;
the design doc is the durable reference.

**Tests**: 60+ new unit tests across `receiver_confirm_test`,
`receiver_budget_test`, `transfer_waiter_test`, and the F3-1 hardening
suite — skew matrix, kill-switch both sides, receiver-truth-wins, retry
healing, mixed fleets, tombstone interplay, unsolicited-confirm guard,
bounded lost-confirm, multi-ref acquisition, quorum intersection, waiter
never-hangs invariants. Full streaming+fobs+legacy sweeps green.
Reviewed by adversarial multi-agent workflows (24 + 6 confirmed findings
fixed) plus a 4-angle cleanup pass (net −48 LOC, hot-path trims).

---------

Co-authored-by: Zhihong Zhang <100308595+nvidianz@users.noreply.github.com>